package com.reger.l2cache.pipeline.core;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisConnectionUtils;
import org.springframework.util.Assert;

public class RedisPipelineCore implements PipelineCore {

	private static final Logger log = LoggerFactory.getLogger(RedisPipelineCore.class);

	private static final int MAX_PIP_SIZE = 500;
	private static final int WARN_TIME_MILLIS = 50;
	private static final int THREAD_NUM = 2 * Runtime.getRuntime().availableProcessors();
	private static final Map<RedisConnectionFactory, RedisPipelineCore> MAP_REDIS_PIPELINE_CORE = new HashMap<>();
	private static final ExecutorService executorService = Executors.newCachedThreadPool(target -> {
		Thread thread = new Thread(target);
		thread.setName("Redis-Pipeline-Work-" + thread.getId());
		thread.setPriority(Thread.MAX_PRIORITY);
		return thread;
	});

	private final LinkedBlockingQueue<QueueNode<?>> queue = new LinkedBlockingQueue<>();
	private final Lock LOCK = new ReentrantLock();
	
	private volatile boolean pipeEnabled = true;
	private volatile int workThreadNum = THREAD_NUM;
	
	private RedisConnectionFactory connectionFactory;

	protected RedisPipelineCore() {
		super();
	}


	public boolean isPipeEnabled() {
		return pipeEnabled;
	}

	public void setPipeEnabled(Boolean pipeEnabled) {
		if(pipeEnabled!=null) {
			this.pipeEnabled = pipeEnabled;
		}
	}

	public RedisConnectionFactory getConnectionFactory() {
		return connectionFactory;
	}

	public void setConnectionFactory(RedisConnectionFactory connectionFactory) {
		this.connectionFactory = connectionFactory;
	}

	public LinkedBlockingQueue<QueueNode<?>> getQueue() {
		return queue;
	}

	public int getWorkThreadNum() {
		return workThreadNum;
	}

	public void setWorkThreadNum(Integer workThreadNum) {
		if(workThreadNum!=null) {
			this.workThreadNum = workThreadNum;
		}
	}


	@Override
	public <T> T execute(Operations operations, Callback<T> callback, boolean relust) {
		if (connectionFactory == null || !pipeEnabled) {
			log.debug("未启用RedisPipeline");
			return this.asyncExecute(operations, callback);
		}
		if (!relust) {
			log.debug("操作可能没有返回值值暂不支持在管道中执行");
			return this.asyncExecute(operations, callback);
		}
		return this.syncExecute(operations, callback);
	}

	/**
	 * 开启管道操作
	 * 
	 * @param connectionFactory
	 * @param pipeEnabled
	 * @param workThreadNum
	 * @return
	 */
	public static synchronized final RedisPipelineCore start(RedisConnectionFactory connectionFactory,
			Boolean pipeEnabled, Integer workThreadNum) {
		Assert.notNull(connectionFactory, "redis链接工厂不可以为空");
		if (MAP_REDIS_PIPELINE_CORE.containsKey(connectionFactory)) {
			return MAP_REDIS_PIPELINE_CORE.get(connectionFactory);
		}
		RedisPipelineCore pipelineCore = new RedisPipelineCore();
		pipelineCore.setConnectionFactory(connectionFactory);
		pipelineCore.setPipeEnabled(pipeEnabled);
		pipelineCore.setWorkThreadNum(workThreadNum);
		pipelineCore.start();
		MAP_REDIS_PIPELINE_CORE.put(connectionFactory, pipelineCore);
		return pipelineCore;
	}

	
	protected LinkedList<QueueNode<?>> getQueueNodes() {
		try {
			LOCK.lockInterruptibly();
		} catch (InterruptedException e1) {
			Thread.currentThread().interrupt();
			log.debug("任务线程被中断", e1);
			return null;
		}
		LinkedList<QueueNode<?>> queueNodes = new LinkedList<>();
		try {
			QueueNode<?> queueNode = null;
			try {
				queueNode = queue.take();
			} catch (InterruptedException e) {
				Thread.currentThread().interrupt();
				log.debug("任务线程被中断", e);
			}
			if (queueNode == null) {
				return null;
			}
			queueNodes.add(queueNode);
			while (queueNodes.size() < MAX_PIP_SIZE && (queueNode = queue.poll()) != null) {
				queueNodes.add(queueNode);
			}
		} finally {
			LOCK.unlock();
		}
		return queueNodes;
	}

	protected void handle() {
		LinkedList<QueueNode<?>> queueNodes = getQueueNodes();
		if (queueNodes == null || queueNodes.isEmpty()) {
			return;
		}
		this.operation(queueNodes);
	}

	protected void operation(List<QueueNode<?>> queueNodes) {
		RedisConnection _connection = null;
		try {
			long start = System.currentTimeMillis();
			try {
				RedisConnection connection = RedisConnectionUtils.getConnection(connectionFactory);
				_connection = connection;
				connection.openPipeline();
				queueNodes.forEach(qn -> qn.operation(connection));
				List<Object> reluts = connection.closePipeline();
				for (int i = 0; i < queueNodes.size(); i++) {
					QueueNode<?> queueNode = queueNodes.get(i);
					Object relut = reluts.get(i);
					queueNode.callback(relut);
				}
			} catch (Exception exception) {
				queueNodes.forEach(queueNode -> queueNode.callback(exception));
			} finally {
				long end = System.currentTimeMillis();
				if ((end - start) > WARN_TIME_MILLIS) {
					log.info("注意是否有redis慢查询，本次执行耗时{}毫秒,{}条指令一次执行", end - start, queueNodes.size());
				}
				RedisConnectionUtils.releaseConnection(_connection, connectionFactory);
			}
		} finally {
			queueNodes.forEach(qn -> qn.aNotify());
		}
	}

	/**
	 * 不在管道中执行
	 * 
	 * @param operations
	 * @param callback
	 * @return
	 */
	protected <T> T asyncExecute(Operations operations, Callback<T> callback) {
		QueueNode<T> queueNode = new QueueNode<>(operations, callback);
		RedisConnection connection = null;
		try {
			connection = RedisConnectionUtils.getConnection(connectionFactory);
			return queueNode.operation(connection).relust();
		} finally {
			RedisConnectionUtils.releaseConnection(connection, connectionFactory);
		}
	}

	/**
	 * 在管道中执行
	 * 
	 * @param operations
	 * @param callback
	 * @return
	 */
	protected <T> T syncExecute(Operations operations, Callback<T> callback) {
		QueueNode<T> queueNode = new QueueNode<>(operations, callback);
		try {
			queue.put(queueNode);
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
			log.warn("加入元素时，线程中断", queueNode);
		}
		return queueNode.await().relust();
	}
	/**
	 * 启用当前管道核心线程操作
	 */
	protected void start() {
		if (workThreadNum == 0) {
			workThreadNum = THREAD_NUM;
		}
		for (int i = 0; i < workThreadNum; i++) {
			executorService.submit(this::startTask);
		}
		log.debug("准备了{}个线程用于处理任务", workThreadNum);
	}
	
	protected void startTask() {
		while (!Thread.interrupted()) {
			try {
				this.handle();
			} catch (Exception e) {
				log.warn("存在异常", e);
			}
		}
		log.info("线程{}结束", Thread.currentThread().getName());
	}

	/**
	 * 关闭管道操作
	 */
	public void stop() {
		Assert.notNull(connectionFactory, "没有开启 RedisPipelineCore 不可以关闭");
		this.connectionFactory = null;
		if (!executorService.isShutdown()) {
			executorService.shutdownNow();
		}
	}

}
